/*
 * Copyright (c) 2012 Robert Burrell Donkin robertburrelldonkin.name
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy of this software 
 * and associated documentation files (the "Software"), to deal in the Software without restriction, 
 * including without limitation the rights to use, copy, modify, merge, publish, distribute, 
 * sublicense, and/or sell copies of the Software, and to permit persons to whom the Software 
 * is furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in all copies or 
 * substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING 
 * BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND 
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, 
 * DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, 
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 * 
 */
package walrus.engine;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import walrus.base.ArticleBuffer;
import walrus.base.ArticleProcessor;
import walrus.base.Constants;

public class ArticleProcessingEngine implements Runnable {

	/** MUST be accessed only by the contained thread */
	private final List<Future<ArticleBuffer>> inFlight;
	private final BlockingQueue<ArticleBuffer> in;
	private final BlockingQueue<ArticleBuffer> out;
	private final ExecutorService executor;
	private final ArticleProcessor processor;

	private final CountDownLatch doneLatch = new CountDownLatch(1);

	private boolean stop = false;
	private boolean finish = false;

	public ArticleProcessingEngine(final ArticleProcessor processor) {
		this(Constants.DEFAULT_NUMBER_OF_BUFFERS, Constants.DEFAULT_NUMBER_OF_THREADS, Executors
				.defaultThreadFactory(), processor);
	}

	public ArticleProcessingEngine(final ArticleProcessor processor, int buffers, final int numberOfThreads) {
		this(buffers, numberOfThreads, Executors
				.defaultThreadFactory(), processor);
	}

	
	public ArticleProcessingEngine(int buffers, final int numberOfThreads,
			final ThreadFactory threadFactory, final ArticleProcessor processor) {
		this(buffers, Executors.newFixedThreadPool(numberOfThreads,
				threadFactory), processor);
	}

	public ArticleProcessingEngine(int buffers, final ExecutorService executor,
			final ArticleProcessor processor) {
		this.executor = executor;
		this.in = new ArrayBlockingQueue<ArticleBuffer>(buffers);
		this.out = new ArrayBlockingQueue<ArticleBuffer>(buffers);
		for (int i = 0; i < buffers; i++) {
			out.add(new ArticleBuffer());
		}
		this.inFlight = new ArrayList<Future<ArticleBuffer>>(buffers);
		this.processor = processor;
	}

	public BlockingQueue<ArticleBuffer> getIn() {
		return in;
	}

	public BlockingQueue<ArticleBuffer> getOut() {
		return out;
	}

	public void run() {
		while (!stop) {
			try {
				final ArticleBuffer incoming = in
						.poll(1, TimeUnit.MILLISECONDS);
				process(incoming);
				forwardCompleted();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (ExecutionException e) {
				e.printStackTrace();
			}
		}
		if (finish) {
			try {
				System.out.println("Tidying up...");
				while (!in.isEmpty()) {
					try {
						process(in.poll());
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				executor.shutdown();
				while (!executor.isTerminated())
					;
				try {
					forwardCompleted();
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (ExecutionException e) {
					e.printStackTrace();
				}
			} finally {
				System.out.println("Done.");
				doneLatch.countDown();
			}
		}
	}

	private void forwardCompleted() throws InterruptedException,
			ExecutionException {
		for (final Iterator<Future<ArticleBuffer>> it = inFlight.iterator(); it
				.hasNext();) {
			final Future<ArticleBuffer> future = it.next();
			if (future.isDone()) {
				final ArticleBuffer buffer = future.get();
				buffer.reset();
				out.put(buffer);
				it.remove();
			}
		}
	}

	private void process(final ArticleBuffer incoming)
			throws InterruptedException {
		if (incoming != null) {
			inFlight.add(executor.submit(newCall(incoming)));
		}
	}

	protected Callable<ArticleBuffer> newCall(final ArticleBuffer incoming) {
		return new ProcessingTask(incoming, processor);
	}

	public void stop() {
		this.stop = true;
	}

	public void finish() {
		stop = true;
		finish = true;
	}

	public void waitTillfinished() throws InterruptedException {
		doneLatch.await();
	}

	@Override
	public String toString() {
		return "ArticleProcessingEngine [doneLatch=" + doneLatch
				+ ", executor=" + executor + ", finish=" + finish + ", in="
				+ in + ", inFlight=" + inFlight + ", out=" + out
				+ ", processor=" + processor + ", stop=" + stop + "]";
	}
}
